package com.alison.source

import org.apache.spark.sql.SparkSession

object E1_std {

  def main(args: Array[String]): Unit = {
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("HelloStructuredStreaming")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    // 创建一个流式DataFrame，这里从Socket读取数据
    val lines = spark.readStream
      .format("socket")
      .option("host", "192.168.56.104")
      .option("port", 9999)
      .load()

    // 对流式DataFrame进行转换，将每一行数据切分成单词
    val words = lines.as[String].flatMap(_.split(" "))

    // 对单词进行分组和聚合
    val wordCounts = words.groupBy("value").count()

    // 启动查询, 把结果打印到控制台
    val query = wordCounts.writeStream
      .outputMode("complete") // 或者选择其他适合的输出模式
//      .outputMode("update") // 或者选择其他适合的输出模式
//      .outputMode("append") // 或者选择其他适合的输出模式,需要配置 watermark 才可以使用 agg 聚合函数
      .format("console")
      .start()

    // 等待应用程序终止
    query.awaitTermination()

    //关闭 Spark
    spark.stop()
  }

}
